package com.zimmor.mq.rocket_train1.transzction;

import com.alibaba.fastjson.JSON;
import com.zimmor.mq.RocketMQProperties;
import io.netty.handler.codec.json.JsonObjectDecoder;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author:zimmor
 * @Date: 2023/2/16 16:47
 * @Description 发送事务消息
 * @version: 1.0.0
 */

public class Producer {
    public static void main(String[] args) throws MQClientException {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("producer-group5");
        transactionMQProducer.setNamesrvAddr(RocketMQProperties.NAME_SRV_ADDR);
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            Map obj = new HashMap();
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                LocalTransactionState state = LocalTransactionState.UNKNOW;
                //获取发送到的消息
//                obj = (Map) JSON.parse(message.getBody());
                try{
                    boolean status = false;
                    if(status) {
                        obj.put("transactionId", message.getTransactionId());
                        state = LocalTransactionState.COMMIT_MESSAGE;
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return state;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                LocalTransactionState state = LocalTransactionState.UNKNOW;
                Object transactionId = obj.get("transactionId");
                try{
                    int i = 1/0;
                    if(transactionId != null) {
                        state = LocalTransactionState.COMMIT_MESSAGE;
                    } else {
                        state = LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }catch (Exception e) {
                    e.printStackTrace();
                }

                return state;
            }
        });
        transactionMQProducer.start();
        Message message = new Message("transaction", "tag", "transactionMessage".getBytes());
        TransactionSendResult transactionSendResult = transactionMQProducer.sendMessageInTransaction(message, null);
        System.out.println("transactionSendResult = " + transactionSendResult);
    }
}
